package com.bwie.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bwie.config.KafkaConfig;
import com.bwie.domain.Log;
import com.bwie.domain.TbRedEnvelope;
import com.bwie.domain.TbUser;
import com.bwie.mapper.LogMapper;
import com.bwie.mapper.TbUserMapper;
import com.bwie.param.TbRedEnvelopeParam;
import com.bwie.service.TbRedEnvelopeService;
import com.bwie.mapper.TbRedEnvelopeMapper;
import com.bwie.utils.R;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.support.master.AcknowledgedRequest;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.redis.connection.ReactiveStreamCommands;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.KafkaListeners;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
* @author xinghe
* @description 针对表【tb_red_envelope(红包表)】的数据库操作Service实现
* @createDate 2025-06-17 15:28:21
*/
@Service
@Slf4j
public class TbRedEnvelopeServiceImpl extends ServiceImpl<TbRedEnvelopeMapper, TbRedEnvelope>
    implements TbRedEnvelopeService{

    @Autowired
    private TbRedEnvelopeMapper tbRedEnvelopeMapper;

    @Autowired
    private TbUserMapper tbUserMapper;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Autowired
    private LogMapper logMapper;

    @KafkaListener(topics = KafkaConfig.KAFKA_STRING)
    public void kafka(String str, Acknowledgment acknowledgment){
        TbRedEnvelopeParam tbRedEnvelopeParam = JSON.parseObject(str, TbRedEnvelopeParam.class);

        TbRedEnvelope tbRedEnvelope = tbRedEnvelopeMapper.selectOne(
                new QueryWrapper<TbRedEnvelope>().lambda().eq(TbRedEnvelope::getRedEnvelopeId, tbRedEnvelopeParam.getRedEnvelopeId())
        );


        tbRedEnvelope.setRemainderPrice(0);
        tbRedEnvelope.setGetIt(1);
        tbRedEnvelopeMapper.updateById(tbRedEnvelope);
        TbUser tbUser = tbUserMapper.selectOne(
                new QueryWrapper<TbUser>().lambda().eq(TbUser::getUserId, tbRedEnvelopeParam.getUserId())
        );
        Integer price1 = tbUser.getPrice();
        tbUser.setPrice(price1+tbRedEnvelope.getPrice());
        tbUserMapper.updateById(tbUser);

        Log log1 = new Log();
        log1.setPrice(tbRedEnvelope.getRemainderPrice());
        log1.setRedId(tbRedEnvelope.getRedEnvelopeId());
        log1.setUserId(tbUser.getUserId());
        log1.setCreateTime(new Date());
        logMapper.insert(log1);
        acknowledgment.acknowledge();
    }

    @Override
    public R redList(TbRedEnvelopeParam tbRedEnvelopeParam) {

        Page<TbRedEnvelope> page = new Page<>(tbRedEnvelopeParam.getPage(), tbRedEnvelopeParam.getLimit());
        Page<TbRedEnvelope> tbRedEnvelopePage = tbRedEnvelopeMapper.selectRedAllPage(page, tbRedEnvelopeParam);
        return R.OK(tbRedEnvelopePage);
    }

    @Override
    public R sendRed(TbRedEnvelopeParam tbRedEnvelopeParam) {

//        BigDecimal price = new BigDecimal(String.valueOf(tbRedEnvelopeParam.getPrice()));
//
//        BigDecimal bigDecimal = new BigDecimal(300.00);

        if(tbRedEnvelopeParam.getPrice() > 300){
        return R.ERROR(500,"不能超过300");
        }

        TbUser tbUser = tbUserMapper.selectOne(
                new QueryWrapper<TbUser>().lambda().eq(TbUser::getUserId, tbRedEnvelopeParam.getUserId())
        );
        Integer price1 = tbUser.getPrice();
        if (tbRedEnvelopeParam.getPrice() > price1) {
            return R.ERROR(500,"账户余额不够");
        }

        if(tbRedEnvelopeParam.getNum() <= 0){
            return R.ERROR(500,"红包个数必须大于0");
        }


        TbRedEnvelope tbRedEnvelope = new TbRedEnvelope();
        tbRedEnvelope.setRedEnvelopeTheme(tbRedEnvelopeParam.getRedEnvelopeTheme());
        tbRedEnvelope.setNum(tbRedEnvelopeParam.getNum());
        tbRedEnvelope.setPrice(tbRedEnvelopeParam.getPrice());
        tbRedEnvelope.setGetIt(0);
        tbRedEnvelope.setStatus(1);
        tbRedEnvelope.setSendTime(new Date());
        tbRedEnvelope.setCollectNum(0);
        tbRedEnvelope.setContinueTime(tbRedEnvelopeParam.getContinueTime());
        tbRedEnvelope.setRemainderPrice(tbRedEnvelopeParam.getPrice());
        tbRedEnvelopeMapper.insert(tbRedEnvelope);


        int price = tbUser.getPrice() - tbRedEnvelopeParam.getPrice();

        tbUser.setPrice(price);
        tbUserMapper.updateById(tbUser);
        return R.OK();
    }

    @Override
    public R robRed(TbRedEnvelopeParam tbRedEnvelopeParam) {

        String lockKey = "rediss-" + tbRedEnvelopeParam.getRedEnvelopeId();

        RLock lock = redissonClient.getLock(lockKey);


        try {
            boolean b = lock.tryLock(5, 10, TimeUnit.MINUTES);
            if(b){
                TbRedEnvelope tbRedEnvelope = tbRedEnvelopeMapper.selectOne(
                        new QueryWrapper<TbRedEnvelope>().lambda().eq(TbRedEnvelope::getRedEnvelopeId, tbRedEnvelopeParam.getRedEnvelopeId())
                );
                kafkaTemplate.send(KafkaConfig.KAFKA_STRING,JSON.toJSONString(tbRedEnvelopeParam));
            }else {
                return R.ERROR(500,"已上锁");
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {

        }



        return R.OK();
//
    }
}




